package cn.jly.flink.utils;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @PackageName cn.jly.flink.utils
 * @ClassName FlinkUtils
 * @Description flink工具类
 * @Author 姬岚洋
 * @Date 2021/1/26 下午 5:03
 */
public class FlinkUtils {
    private FlinkUtils() {
    }

    /**
     * 流执行环境
     *
     * @return
     */
    public static StreamExecutionEnvironment getStreamExecutionEnv() {
        return StreamExecutionEnvironment.getExecutionEnvironment();
    }

    /**
     * 批执行环境
     *
     * @return
     */
    public static ExecutionEnvironment getBatchExecutionEnv() {
        return ExecutionEnvironment.getExecutionEnvironment();
    }

    /**
     * 创建表执行环境
     *
     * @param streamExecutionEnvironment 流执行环境
     * @return
     */
    public static StreamTableEnvironment createStreamTableEnv(StreamExecutionEnvironment streamExecutionEnvironment) {
        return StreamTableEnvironment.create(streamExecutionEnvironment);
    }

    /**
     * 执行流处理
     *
     * @param env
     * @param jobName
     * @throws Exception
     */
    public static void executeStream(StreamExecutionEnvironment env, String jobName) throws Exception {
        if (env == null) {
            return;
        }
        if (jobName == null || jobName.isEmpty()) {
            env.execute();
        } else {
            env.execute(jobName);
        }
    }
}
